package com.lm.flink.log;

import com.lm.flink.entry.TotalCount;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;

public class RedisTotalCountSink extends RichSinkFunction<TotalCount> {
    private transient JedisPool jedisPool;

    @Override
    public void open(Configuration parameters) throws Exception {
        JedisPoolConfig poolConfig = new JedisPoolConfig();
        poolConfig.setMaxTotal(10);
        poolConfig.setMaxIdle(5);
        poolConfig.setMinIdle(1);

        jedisPool = new JedisPool(poolConfig, "10.2.0.230", 6379, 2000, "123456");
    }

    @Override
    public void invoke(TotalCount count, Context context) throws Exception {
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.select(0);

            // 存储总计数
            jedis.set("log:total:count", String.valueOf(count.getTotalCount()));

            // 存储时间序列数据
            jedis.zadd("log:total:timeline", count.getWindowEnd(), String.valueOf(count.getTotalCount()));

            // 更新总览
            jedis.hset("log:stats", "total", String.valueOf(count.getTotalCount()));

            System.out.println("Updated total count: " + count.getTotalCount());
        }
    }

    @Override
    public void close() throws Exception {
        if (jedisPool != null) {
            jedisPool.close();
        }
    }
}
